package com.bw.service;

import com.bw.domain.MyMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.retry.annotation.Retryable;
import org.springframework.stereotype.Service;

/**
 * 生产者的service    如果你在controller 需要 发送消息    就调用这个service
 */
@Service
@Slf4j
public class SenderService {

    @Autowired
    RocketMQTemplate rocketMQTemplate;

    /**
     * 生产者 发消息
     * 同步发送
     *
     * @param message
     */
    @Retryable  // spring的重试注解， 如果该程序抛异常 默认重试执行3次，  （（生产者如果发送失败。也会重试3次））
    public void sendMsg(MyMessage message) {
        SendResult topic = rocketMQTemplate.syncSend("sms_topic", message);//   同步   成功的把消息放入 队列中
        //  消息放入 队列成功
        if (SendStatus.SEND_OK.equals(topic.getSendStatus())) {
            log.info("生产者 发送成功！！！！！");
        } else {
            log.error("生产者 发送失败！！！！！{}", message);
        }

    }

    /**
     * 异步发送
     *
     * @param message
     */
    @Retryable  // spring的重试注解， 如果该程序抛异常 默认重试执行3次，  （（生产者如果发送失败。也会重试3次））
    public void sendAMsg(MyMessage message) {

        rocketMQTemplate.asyncSend("sms_topic", message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {  // 如果发送成功，这行代码会被触发执行
                System.out.println("发送成功了");
            }

            @Override
            public void onException(Throwable e) {  // 如果发送失败，这行代码会被触发执行
                System.out.println("发送失败了");
            }
        });

    }

    /**
     * 顺序消息: 发送消息时,让具有相同规则的消息，存放到同一个队列中，  消费者在消费时，可以按照顺序消费
     *
     * @param message
     */
    //
    @Retryable
    public void sendOrderly(MyMessage message) {

        SendResult result = rocketMQTemplate.syncSendOrderly("sms_topic", message, "orderMsg");
        if (SendStatus.SEND_OK.equals(result.getSendStatus())) {
            log.info("生产者 发送成功！！！！！");
        } else {
            log.error("生产者 发送失败！！！！！{}", message);
        }
    }


    /**
     * 延时消息  创建订单时， 发送 延时消息， 消费者接收消息时判断订单是否付款，如果没有付款 则 关闭订单
     */
    // 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
    // 发送同步延迟消息
    public void sendOrderClose(String orderId) {
        // 构建消息对象
        Message<String> message = MessageBuilder.withPayload(orderId).build();
        // 参数1:主题  参数2:消息内容 参数3:连接mq的超时时长 参数4:延迟时间等级
        SendResult result = rocketMQTemplate.syncSend("yanshi5", message, 500, 9);
        if (SendStatus.SEND_OK.equals(result.getSendStatus())) {
            log.info("生产者 发送成功！！！！！");
        } else {
            log.error("生产者 发送失败！！！！！{}", message);
        }
    }


}
